package net.bwie.realtime.jtp.dwd.order.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dwd.order.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/06/03 21:27:50
 * @version: 1.0
 */
public class RequestDicDimAsyncFunction extends RichAsyncFunction<String, String> {


    private ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
//        创建线程池
        executorService = Executors.newScheduledThreadPool(10);
    }

    @Override
    public void asyncInvoke(String s, ResultFuture<String> resultFuture) throws Exception {
        // a-解析JSON字符串
        JSONObject jsonObject = JSON.parseObject(s);
        String rowKey = jsonObject.getString("source_type");
        // b-异步请求HBase获取数据
        final Future<String> result = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return HbaseUtil.getData("dim_base_dic", rowKey, "info", "dic_name");
            }
        });

        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    String requestValue = result.get();
                    // 添加字段值
                    jsonObject.put("source_type_name", requestValue);
                } catch (InterruptedException | ExecutionException e) {
                    // 设置字段为：exception
                    jsonObject.put("source_type_name", "exception");
                }
                // 返回值
                return jsonObject.toString();
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(dbResult));
        });

    }

    @Override
    public void timeout(String input, ResultFuture<String> resultFuture) throws Exception {
        // 解析JSON
        JSONObject jsonObject = JSON.parseObject(input);
        // 设置字段为：timeout
        jsonObject.put("source_type_name", "timeout");
        // 返回
        resultFuture.complete(Collections.singleton(jsonObject.toString()));
    }

    @Override
    public void close() throws Exception {
        executorService.shutdownNow();
    }


}
